package com.atguigu.gmall;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.gmall.utils.Constant.KafkaConstant;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;

/**
 * @ClassName: MyCanalClient
 * @Description:
 * @Author: kele
 * @Date: 2021/3/20 16:55
 *
 * 需求：统计每天新增订单中的final_amount进行累加求和
 *
 *
 * 1、连接canalserver
 * 2、订阅canal server中的binlog数据
 * 3、拉取数据
 * 4、从拉取的数据中筛选order_info表中insert的数据
 * 5、封装数据为json格式
 * 6、将数据写入kafka中
 *
 **/
public class MyCanalClient {

    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {

        /**
         * 1、配置连接
         *
         * newSingleConnector
         *
         * 参数列表：
         *      SocketAddress address：canal运行的主机名和端口号
         *      String destination：订阅的mysql的目标名，在canal的配置中，
         *      String username：
         *      String password：
         */
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(

                new InetSocketAddress("hadoop102", 11111),
                "example",
                "",
                ""
        );


        /**
         * 2、获取连接
         */

        canalConnector.connect();

        /**
         * 3、客户端订阅
         *
         * 如果不设置，会按照默认server的配置为准，配置白名单（默认全部读取），黑名单
         *
         * 订阅d1026库下的order_info表
         *
         */

        canalConnector.subscribe("d1026.*67");



        while(true){

            /**
             * 4、获取数据：message：一批数据，包含多个entry
             *
             *      entry：一个entry造成的sql变化
             *
             * get(10)：一次最多拉取10条，少于10条也会拉取，不阻塞
             *
             * 流式处理，需要死循环一致运行
             */
            Message message = canalConnector.get(10);

            //如果没有数据就暂停
            if(message.getId() == -1){

                Thread.sleep(3000);
                continue;

            }
            /**每个message中包含多个entry，遍历
             * entry中有：
             *      1、Header:属性信息，包含语句影响的表名
             *      2、EntryType
             *      3、Store Value
             *      4、RowChange
             *          1、EventType：包含行变化的类型：insert delete update
             *          2、RowDataList：包含Rowdata
             *
             */

            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {

                //EntryType包含多种，有事务开启的，事务关闭，事务回滚，行数据变化的Rowdata
                //这里只需要行变化的数据
                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)) {

                    //影响的数据
                    ByteString storeValue = entry.getStoreValue();

                    //包含表名，数据类型
                    String tableName = entry.getHeader().getTableName();

                    //处理数据，对数据进行过滤，并将数据作为字符串的方式返回，便于后续将数据存入kafka中
                    handler(storeValue,tableName);

                }


            }
        }



    }

    /**
     *
     * @param storeValue
     * @param tableName
     * @throws InvalidProtocolBufferException
     *
     * 处理数据
     * 1、根据需求，需要处理order_onfo的表
     * 2、判断order_info表中insert的操作，
     * 3、sql语句之前是序列化的，需要反序列化，获取反序列化的每行的sql语句
     * 4、一个sql的insert可能影响多行数据，需要遍历每一行的数据
     * 5、将每行的数据封装json格式
     * 6、将数据写入kafka中
     */

    private static void handler(ByteString storeValue, String tableName) throws InvalidProtocolBufferException {

        //只是针对于order_info的数据进行数据过滤
        if ("order_info".equals(tableName)) {

            //获取每行反序列化后的数据(即每个sql的语句),一个sql可能影响多行的数据，需要遍历每一行的数据
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

            //获取每行数据的类型，只保留insert的数据
            if (rowChange.getEventType().equals(CanalEntry.EventType.INSERT)) {

                //
                List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

                //获取的每行数据
                for (CanalEntry.RowData rowData : rowDatasList) {

                    JSONObject jsonObject = new JSONObject();

                    //获取每行数据的每一列
                    for (CanalEntry.Column column : rowData.getAfterColumnsList()) {

                        jsonObject.put(column.getName(), column.getValue());
                    }

                    //将数据写入kafka中时
                    MyProducer.send(KafkaConstant.KAFKA_TOPIC_ORDER_INFO,jsonObject.toString());
                }
            }
        }

    }
}
